/*   
 * Project: simple-rpc-zookeeper
 * FileName: ZookeeperServiceImpl.java
 * Company: Chengdu Sicent Technology Co.,Ltd
 * version: V1.0
 */
package org.simple.rpc.zookeeper.service.impl;

import java.util.ArrayList;
import java.util.List;

import javax.annotation.PostConstruct;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.EnsurePath;
import org.apache.zookeeper.CreateMode;
import org.simple.rpc.zookeeper.service.ZookeeperService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

/**
 * Description:
 * 
 * @author: wangbo
 * @date: 2015年9月2日下午1:51:21
 */

@Component
public class ZookeeperServiceImpl implements ZookeeperService {

	private static CuratorFramework client;
	@Autowired
	@Qualifier("zkUrl")
	private String url;
	@Autowired
	@Qualifier("nameSpace")
	private String nameSpace;

	public String getNameSpace() {
		return nameSpace;
	}

	public void setNameSpace(String nameSpace) {
		this.nameSpace = nameSpace;
	}

	public String getUrl() {
		return url;
	}

	public void setUrl(String url) {
		this.url = url;
	}

	@PostConstruct
	public void init() {
		ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000,
				3);
		client = CuratorFrameworkFactory.newClient(url, retryPolicy);
		client.start();
		EnsurePath ensurePath = client.newNamespaceAwareEnsurePath(nameSpace);
		try {
			ensurePath.ensure(client.getZookeeperClient());
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public void createNode(String path) {
		try {
			client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path);
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

	public void createEpheMeralNode(String path) {
		try {
			client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path);
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

	public void setNodeData(String path, String data) {
		try {
			client.setData().forPath(path, data.getBytes());
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

	public void deleteNode(String path) {
		try {
			client.delete().forPath(path);
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

	public List<String> getChilds(String path) {
		try {
			return client.getChildren().forPath(path);
		} catch (Exception e) {
			e.printStackTrace();
			return null;
		}
	}

	public boolean isConnected() {
		return client.getZookeeperClient().isConnected();
	}

	public void close() {
		client.close();
	}

	/* (non-Javadoc)
	 * @see org.simple.rpc.zookeeper.service.ZookeeperService#getData(java.lang.String)
	 */
	public byte[] getData(String path) {
		try {
			return client.getData().forPath(path);
		} catch (Exception e) {
			e.printStackTrace();
			return null;
		}
	}

	public PathChildrenCache pathChildrenCache(
			String path, Boolean cacheData) {
		  final PathChildrenCache cached = new PathChildrenCache(client, path, cacheData);
		  try {
			cached.start(StartMode.POST_INITIALIZED_EVENT);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	        cached.getListenable().addListener(new PathChildrenCacheListener() { 
	            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
	                PathChildrenCacheEvent.Type eventType = event.getType();
	                System.out.println(eventType.toString()+"===========================");
	                switch (eventType) {
	                    case CHILD_ADDED:
	                    	 System.out.println("add......");
	                    	 break;
	                    case CHILD_REMOVED:
	                    	 System.out.println("remove....");
	                    	 break;
	                    case CHILD_UPDATED:
	                    	System.out.println("update....");
	                    	break;
	                    default:
	                        System.out.println("PathChildrenCache changed : {path:" + event.getData().getPath() + " data:" +
	                                new String(event.getData().getData()) + "}");
	                }
	            }
	        });
	        return cached;
	}

	/* (non-Javadoc)
	 * @see org.simple.rpc.zookeeper.service.ZookeeperService#nodeCache(java.lang.String, java.lang.Boolean)
	 */
	public void nodeCache(final String path, Boolean dataIsCompressed) {
		// TODO Auto-generated method stub
		final NodeCache cache=new NodeCache(client, path, dataIsCompressed);
		try {
			cache.start();
			cache.getListenable().addListener(new NodeCacheListener() {
				public void nodeChanged() throws Exception {
					System.out.println("path:"+path+"has new data============>>>>>>>>");
					System.out.println(cache.getCurrentData().getData());
				}
			});
		} catch (Exception e) {
			e.printStackTrace();
		}
		
	}

	/* (non-Javadoc)
	 * @see org.simple.rpc.zookeeper.service.ZookeeperService#getChildDatas(java.lang.String, java.util.List)
	 */
	public List<String> getChildDatas(String basePath, List<String> childNodes) {
		List<String> list=new ArrayList<String>();
		for (String node : childNodes) {
			String path=basePath+"/"+node;
			byte[] bytes= getData(path);
			list.add(new String(bytes));
		}
		return list;
	}

}
